package com.example.demo.service;

import com.example.demo.entity.UserLogInfo;
import com.example.demo.serializer.UserLogSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaTemplate<String, UserLogInfo> template;

    @Override
    public String send(UserLogInfo userLogInfo) {

        final ListenableFuture<SendResult<String, UserLogInfo>> future = template
                .send("user-log-action", userLogInfo.getAction(), userLogInfo);
        // 同步发送消息
        try {
            final SendResult<String, UserLogInfo> sendResult = future.get();
            final RecordMetadata metadata = sendResult.getRecordMetadata();

            System.out.println("消息请发送成功:  " + "\t" +
                    "主题为：" + metadata.topic() + "\t" +
                    "分区：" + metadata.partition() + "\t" +
                    "偏移量：" + metadata.offset());

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return "ok";
    }
}
